package tail

import (
	"context"
	"logAgent/kafka"
	"logAgent/utils"
	"time"

	"github.com/hpcloud/tail"
)

//TailTask 监控任务
type TailTask struct {
	path   string //日志路径
	topic  string //日志要传递到kafka那个topic
	child  *tail.Tail
	ctx    context.Context
	cancel context.CancelFunc
}

//newTailTask 新建一个监控任务
func newTailTask(path, topic string) *TailTask {
	ctx, cancel := context.WithCancel(context.Background())
	tailObj := TailTask{
		path:   path,
		topic:  topic,
		ctx:    ctx,
		cancel: cancel,
	}
	err := tailObj.init()
	if err != nil {
		utils.Log.Errorf("Path: %s, Topic: %s, 新建监控任务失败: %v\n", path, topic, err)
		return nil
	}
	return &tailObj
}

//init 监控任务初始化
func (t *TailTask) init() error {
	//设置监控配置
	config := tail.Config{
		ReOpen:    true,                                 //重新打开
		Follow:    true,                                 //继续找下一行
		Location:  &tail.SeekInfo{Offset: 0, Whence: 2}, //定位
		MustExist: false,                                //不存在则跳过
		Poll:      true,                                 //轮询
	}
	//连接
	child, err := tail.TailFile(t.path, config)
	if err != nil {
		utils.Log.Errorf("path=%s的文件监控失败: %v\n", t.path, err)
		return err
	}
	t.child = child
	//执行任务
	go t.run()
	return nil
}

//run 执行任务
func (t *TailTask) run() {
	for {
		select {
		case line := <-t.child.Lines:
			kafka.SendToChan(t.topic, line.Text) //发送信息到管道
		case <-t.ctx.Done(): //接收中断信息
			utils.Log.Infof("Path: %s, Topic: %s 的任务正在中断\n", t.path, t.topic)
			return
		default:
			time.Sleep(time.Second)
		}
	}
}
